package com.ruoyi.flink.connectors;/**
 * @program: ruoyi
 * @description: KafkaConnectorJavaConsumer
 * @author: zengchen
 * @create: 2020-12-15 20:12
 **/

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

/**
 * @program: ruoyi
 * @description: KafkaConnectorJavaConsumer
 * @author: zengchen
 * @create: 2020-12-15 20:12
 **/
public class KafkaConnectorJavaProducer {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "centos7-out:9092");
//        properties.setProperty("group.id", "zc_test");
//        FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer("study_test", new SimpleStringSchema(), properties);

        DataStreamSource<String> localhost = env.socketTextStream("localhost", 9999);

        localhost.print();

//        FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer("study_test", (KafkaSerializationSchema) new SimpleStringSchema(),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer<String>("study_test", new SimpleStringSchema(), properties);
        localhost.addSink(flinkKafkaProducer);


        env.execute("KafkaConnectorJavaConsumer");
    }
}
